package com.thunderbit.kafka.producer;

import com.thunderbit.kafka.config.KafkaConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;
import java.util.concurrent.Future;

/**
 * Kafka 简单生产者示例
 * 演示如何发送消息到 Kafka
 */
public class SimpleProducer {
    
    private static final Logger logger = LoggerFactory.getLogger(SimpleProducer.class);
    
    public static void main(String[] args) {
        // 1. 获取配置
        Properties props = KafkaConfig.getProducerProperties();
        
        // 2. 创建 Producer 实例
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        
        try {
            // 3. 发送 10 条消息
            for (int i = 0; i < 10; i++) {
                String key = "key-" + i;
                String value = "Hello Kafka! Message number: " + i;
                
                // 创建 ProducerRecord
                ProducerRecord<String, String> record = 
                    new ProducerRecord<>(KafkaConfig.TOPIC_NAME, key, value);
                
                // 发送消息（异步）
                Future<RecordMetadata> future = producer.send(record, (metadata, exception) -> {
                    if (exception == null) {
                        // 发送成功
                        logger.info("消息发送成功 - Topic: {}, Partition: {}, Offset: {}, Key: {}", 
                                    metadata.topic(), 
                                    metadata.partition(), 
                                    metadata.offset(),
                                    key);
                    } else {
                        // 发送失败
                        logger.error("消息发送失败 - Key: {}, Error: {}", key, exception.getMessage());
                    }
                });
                
                // 可选：等待发送完成（同步发送）
                // RecordMetadata metadata = future.get();
                
                // 模拟延迟
                Thread.sleep(100);
            }
            
            logger.info("所有消息已发送");
            
        } catch (Exception e) {
            logger.error("发送消息时发生错误", e);
        } finally {
            // 4. 关闭 Producer（会等待所有消息发送完成）
            producer.close();
            logger.info("Producer 已关闭");
        }
    }
}